#include <stdio.h>
#include <signal.h>
#include "kvs.h"
#include "master_slave.h"
#include <stdatomic.h>
#include "public.h"

// ==================================JEMALLOC==========================================
#include <jemalloc.h>

#ifdef JEMALLOC_MANGLE // 部分jemalloc版本会自动重命名函数（加前缀）
#define jemalloc_malloc je_malloc
#define jemalloc_free je_free
#else
#define jemalloc_malloc malloc
#define jemalloc_free free
#endif


void *kvs_malloc(size_t size) {
    if (size == 0) return NULL;
    return jemalloc_malloc(size);
}

void kvs_free(void *ptr) {
    if (ptr == NULL) return;
    jemalloc_free(ptr);
}


// jemalloc轻量统计函数
void kvs_jemalloc_print_stats()
{
    printf("\n===== Jemalloc 内存池状态（3-4线程场景）=====\n");
    malloc_stats_print(NULL, NULL, NULL);  // 先验证这个函数是否能正常调用（无报错则说明统计功能已开启）

    
#ifdef JEMALLOC_MANGLE
#define JEMALLOC_MALLCTL je_mallctl  // 定义带 je_ 前缀的宏

#else
#define JEMALLOC_MALLCTL mallctl     // 定义不带前缀的宏

#endif

    size_t allocated = 0, active = 0, tcache_hit = 0, tcache_miss = 0;

    if (JEMALLOC_MALLCTL("stats.allocated", &allocated, NULL, NULL, 0) != 0) {
        printf("警告：获取 stats.allocated 失败\n");
    }
    if (JEMALLOC_MALLCTL("stats.active", &active, NULL, NULL, 0) != 0) {
        printf("警告：获取 stats.active 失败\n");
    }
    if (JEMALLOC_MALLCTL("tcache.hit", &tcache_hit, NULL, NULL, 0) != 0) {
        printf("警告：获取 tcache.hit 失败\n");
    }
    if (JEMALLOC_MALLCTL("tcache.miss", &tcache_miss, NULL, NULL, 0) != 0) {
        printf("警告：获取 tcache.miss 失败\n");
    }


    double frag = (active == 0) ? 0.0 : (double)(active - allocated) / active * 100;
    double tcache_ratio = (tcache_hit + tcache_miss == 0) ? 100.0 : (double)tcache_hit / (tcache_hit + tcache_miss) * 100;

    
    printf("已分配内存: %zu KB | 活跃内存: %zu KB\n", allocated / 1024, active / 1024);
    printf("内存碎片率: %.2f%%（<15%% 为优秀）\n", frag);
    printf("TCache 命中率: %.2f%%（>70%% 为充分利用）\n", tcache_ratio);
    printf("============================================\n\n");

    // 取消宏定义（避免影响其他代码）
#undef JEMALLOC_MALLCTL
}
// =====================================================================================




// 主从模式全局变量
#if MASTER_SLAVE
ServerMode g_server_mode = MODE_STANDALONE;
char g_master_ip[32] = {0};
int g_master_port = 0;
#endif

#if MASTER_SLAVE
_Atomic(off_t) aof_file_sync_length; // 主机查看AOF文件大小
#endif

AOFAsyncWriter *global_aof_writer = NULL; // 日志异步写入器




#define MAX_TOKENS 4 // 单条命令最大分包数
// 选择key跟value的分隔方式，1 --> ' '分隔（key不能含空格，value可以）， 0 --> '\x02'分隔（两者都可以含空格）
#define ONLY_VALUE_HAS_SPACE 1

rbtree_t *global_rbtree = NULL;
hash_t *global_hash_table = NULL;
skiplist_t *global_skiplist = NULL;

const char *command[] =
    {
        "RSET",
        "RGET",
        "RDEL",
        "RDELA",
        "RMOD",
        "REXIST",
        "RCOUNT",

        "HSET",
        "HGET",
        "HDEL",
        "HDELA",
        "HMOD",
        "HEXIST",
        "HCOUNT",

        "SSET",
        "SGET",
        "SDEL",
        "SDELA",
        "SMOD",
        "SEXIST",
        "SCOUNT",

        "END"};

enum
{
    KVS_CMD_START = 0,

    // rbtree
    KVS_CMD_RSET = KVS_CMD_START,
    KVS_CMD_RGET,
    KVS_CMD_RDEL,
    KVS_CMD_RDELA,
    KVS_CMD_RMOD,
    KVS_CMD_REXIST,
    KVS_CMD_RCOUNT,
    // hash
    KVS_CMD_HSET,
    KVS_CMD_HGET,
    KVS_CMD_HDEL,
    KVS_CMD_HDELA,
    KVS_CMD_HMOD,
    KVS_CMD_HEXIST,
    KVS_CMD_HCOUNT,
    // skiplist
    KVS_CMD_SSET,
    KVS_CMD_SGET,
    KVS_CMD_SDEL,
    KVS_CMD_SDELA,
    KVS_CMD_SMOD,
    KVS_CMD_SEXIST,
    KVS_CMD_SCOUNT,

    END
};

int kvs_split_token(char *msg, char *tokens[])
{

#if ONLY_VALUE_HAS_SPACE
    int idx = 0;
    int msg_len = strlen(msg);
    char *start = msg;
    // LOG("msg: %s, msg_len: %d\n", msg, msg_len);
    char *token = strtok(msg, " ");
    int segment_len = 10;

    // key不能含空格，value可以含空格，当然也可以规定协议（比如SET key value这三个字符以别的方式分隔，而不是空格。这样key和value就都可以含空格了）。
    if (token != NULL)
    {
        if (strcmp(token, "RSET") == 0 || strcmp(token, "HSET") == 0 || strcmp(token, "SSET") == 0 ||
            strcmp(token, "RMOD") == 0 || strcmp(token, "HMOD") == 0 || strcmp(token, "SMOD") == 0)
        {
            segment_len = 3;
        }
    }

    while (token != NULL)
    {

        // LOG("idx[%d]: %s\n", idx, token);
        tokens[idx++] = token;
        segment_len--;

        if (segment_len == 0)
        {
            // value后面如果有空格的话置为'\0';
            if (strlen(tokens[0]) + strlen(tokens[1]) + strlen(tokens[2]) != msg_len)
            {
                // 说明key中含有空格且第一个空格被置为了'\0', 这里将'\0'置回' '
                *(tokens[2] + strlen(tokens[2])) = ' ';
                // 确保token[2]以'\0'结尾
                *(start + msg_len) = '\0';
            }
            break;
        }
        token = strtok(NULL, " ");
    }
#else
    int idx = 0;
    char *token = strtok(msg, "\x02");

    // key不能含空格，value可以含空格，当然也可以规定协议（比如SET key value这三个字符以别的方式分隔，而不是空格。这样key和value就都可以含空格了）。
    while (token != NULL)
    {

        // LOG("idx[%d]: %s\n", idx, token);
        tokens[idx++] = token;
        token = strtok(NULL, "\x02");
    }

#endif
    /*
    假设原始 msg 是字符串 "SET key val"（内存中存储为 'S','E','T',' ','k','e','y',' ','v','a','l','\0'）：
    第一次调用 strtok(msg, " ") 时，strtok 会找到第一个空格（位于索引 3 的位置），并将这个空格直接替换为 '\0'。此时 msg 的内存变成：'S','E','T','\0','k','e','y',' ','v','a','l','\0'。
    函数返回指向第一个字符 'S' 的指针（即 tokens[0] 指向这里），此时 tokens[0] 对应的字符串是 "SET"（因为遇到 '\0' 终止）。
    第二次调用 strtok(NULL, " ") 时，strtok 从上次替换的 '\0' 位置（索引 3）继续扫描，找到下一个空格（索引 7），再将其替换为 '\0'。msg 内存变成：'S','E','T','\0','k','e','y','\0','v','a','l','\0'。
    函数返回指向 'k' 的指针（tokens[1] 指向这里），对应字符串 "key"。
    第三次调用 strtok(NULL, " ") 时，扫描到字符串末尾的自然 '\0'，返回指向 'v' 的指针（tokens[2] 指向这里），对应字符串 "val"。
    */

    return idx;
}

int kvs_parse_protocol(content_node *wlist, char *tokens[], int count)
{
    if (wlist == NULL || tokens[0] == NULL || count == 0)
        return -1;

    int cmd = KVS_CMD_START;

    // LOG("tokens[0]: %s\n", tokens[0]);
    // LOG("tokens[0]: %s, tokens[1]: %s, tokens[2]: %s\n", tokens[0], tokens[1], tokens[2]);
    for (cmd = KVS_CMD_START; cmd < END; cmd++)
    {
        if (strcmp(command[cmd], tokens[0]) == 0)
        {
            break;
        }
    }

    char *key = tokens[1];
    char *value = tokens[2];

    // LOG("parse_protol(%d): %s\n", cmd, command[cmd]);

    switch (cmd)
    {
    // rbtree
    case KVS_CMD_RSET:
    {

        int ret = kvs_rbtree_set(global_rbtree, key, value);
        if (ret == 0)
        {
            insert_list("RSET SUCCESS\r\n", strlen("RSET SUCCESS\r\n"), wlist);
        }
        else if (ret == 1)
        {
            insert_list("REXIST\r\n", strlen("REXIST\r\n"), wlist);
        }
        else if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_RGET:
    {
        char *buf = kvs_rbtree_get(global_rbtree, key);
        if (buf != NULL)
        {
            insert_list(buf, strlen(buf), wlist);
        }

        else if (buf == NULL)
        {
            insert_list("NO REXIST\r\n", strlen("NO REXIST\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_RDEL:
    {
        int ret = kvs_rbtree_del(global_rbtree, key);
        if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }
        else if (ret == 0)
        {
            insert_list("RDEL SUCCESS\r\n", strlen("RDEL SUCCESS\r\n"), wlist);
        }
        else if (ret > 0)
        {
            insert_list("NO REXIST\r\n", strlen("NO REXIST\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_RDELA:
    {
        int ret = kvs_rbtree_all_del(global_rbtree);
        if (ret == 0)
        {
            insert_list("RDELA SUCCESS\r\n", strlen("RDELA SUCCESS\r\n"), wlist);
        }
        else if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_RMOD:
    {
        int ret = kvs_rbtree_mod(global_rbtree, key, value);
        if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }
        else if (ret == 0)
        {
            insert_list("RMOD SUCCESS\r\n", strlen("RMOD SUCCESS\r\n"), wlist);
        }
        else if (ret > 0)
        {
            insert_list("NO REXIST\r\n", strlen("NO REXIST\r\n"), wlist);
        }
        break;
    }

    case KVS_CMD_REXIST:
    {
        int ret = kvs_rbtree_exist(global_rbtree, key);
        if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }

        if (ret == 1)
        {
            insert_list("NO REXIST\r\n", strlen("NO REXIST\r\n"), wlist);
        }
        else if (ret == 0)
        {
            insert_list("REXIST\r\n", strlen("REXIST\r\n"), wlist);
        }
        break;
    }

    case KVS_CMD_RCOUNT:
    {
        int count = kvs_rbtree_count(global_rbtree);
        if (count == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }
        else
        {
            char buffer[BUFFER_LENGTH] = {0};
            sprintf(buffer, "RCOUNT: %d\r\n", count);
            insert_list(buffer, strlen(buffer), wlist);
        }

        break;
    }

    // hash
    case KVS_CMD_HSET:
    {
        int ret = kvs_hash_set(global_hash_table, key, value);
        if (ret == 0)
        {
            insert_list("HSET SUCCESS\r\n", strlen("HSET SUCCESS\r\n"), wlist);
        }
        else if (ret == 1)
        {
            insert_list("HEXIST\r\n", strlen("HEXIST\r\n"), wlist);
        }
        else if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_HGET:
    {
        char *buf = kvs_hash_get(global_hash_table, key);
        if (buf != NULL)
        {
            insert_list(buf, strlen(buf), wlist);
        }
        else if (buf == NULL)
        {
            insert_list("NO HEXIST\r\n", strlen("NO HEXIST\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_HDEL:
    {
        int ret = kvs_hash_del(global_hash_table, key);
        if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }
        else if (ret == 0)
        {
            insert_list("HDEL SUCCESS\r\n", strlen("HDEL SUCCESS\r\n"), wlist);
        }
        else if (ret > 0)
        {
            insert_list("NO HEXIST\r\n", strlen("NO HEXIST\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_HDELA:
    {
        int ret = kvs_hash_all_del(global_hash_table);
        if (ret == 0)
        {
            insert_list("HDELA SUCCESS\r\n", strlen("HDELA SUCCESS\r\n"), wlist);
        }
        else if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_HMOD:
    {
        int ret = kvs_hash_mod(global_hash_table, key, value);
        if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }
        else if (ret == 0)
        {
            insert_list("HMOD SUCCESS\r\n", strlen("HMOD SUCCESS\r\n"), wlist);
        }
        else if (ret > 0)
        {
            insert_list("NO HEXIST\r\n", strlen("NO HEXIST\r\n"), wlist);
        }
        break;
    }

    case KVS_CMD_HEXIST:
    {
        int ret = kvs_hash_exist(global_hash_table, key);
        if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }

        if (ret == 1)
        {
            insert_list("NO HEXIST\r\n", strlen("NO HEXIST\r\n"), wlist);
        }
        else if (ret == 0)
        {
            insert_list("HEXIST\r\n", strlen("HEXIST\r\n"), wlist);
        }
        break;
    }

    case KVS_CMD_HCOUNT:
    {
        int count = kvs_hash_count(global_hash_table);
        if (count == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }
        else
        {
            char buffer[BUFFER_LENGTH] = {0};
            sprintf(buffer, "HCOUNT: %d\r\n", count);
            insert_list(buffer, strlen(buffer), wlist);
        }

        break;
    }

    // skiplist
    case KVS_CMD_SSET:
    {

        int ret = kvs_skiplist_set(global_skiplist, key, value);
        if (ret == 0)
        {
            insert_list("SSET SUCCESS\r\n", strlen("SSET SUCCESS\r\n"), wlist);
        }
        else if (ret == 1)
        {
            insert_list("SEXIST\r\n", strlen("SEXIST\r\n"), wlist);
        }
        else if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_SGET:
    {
        char *buf = kvs_skiplist_get(global_skiplist, key);
        if (buf != NULL)
        {
            insert_list(buf, strlen(buf), wlist);
        }
        else if (buf == NULL)
        {
            insert_list("NO SEXIST\r\n", strlen("NO SEXIST\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_SDEL:
    {
        int ret = kvs_skiplist_del(global_skiplist, key);
        if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }
        else if (ret == 0)
        {
            insert_list("SDEL SUCCESS\r\n", strlen("SDEL SUCCESS\r\n"), wlist);
        }
        else if (ret > 0)
        {
            insert_list("NO SEXIST\r\n", strlen("NO SEXIST\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_SDELA:
    {
        int ret = kvs_skiplist_all_del(global_skiplist);
        if (ret == 0)
        {
            insert_list("SDELA SUCCESS\r\n", strlen("SDELA SUCCESS\r\n"), wlist);
        }
        else if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }

        break;
    }

    case KVS_CMD_SMOD:
    {
        int ret = kvs_skiplist_mod(global_skiplist, key, value);
        if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }
        else if (ret == 0)
        {
            insert_list("SMOD SUCCESS\r\n", strlen("SMOD SUCCESS\r\n"), wlist);
        }
        else if (ret > 0)
        {
            insert_list("NO SEXIST\r\n", strlen("NO SEXIST\r\n"), wlist);
        }
        break;
    }

    case KVS_CMD_SEXIST:
    {
        int ret = kvs_skiplist_exist(global_skiplist, key);
        if (ret == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }

        if (ret == 1)
        {
            insert_list("NO SEXIST\r\n", strlen("NO SEXIST\r\n"), wlist);
        }
        else if (ret == 0)
        {
            insert_list("SEXIST\r\n", strlen("SEXIST\r\n"), wlist);
        }
        break;
    }

    case KVS_CMD_SCOUNT:
    {
        int count = kvs_skiplist_count(global_skiplist);
        if (count == -1)
        {
            insert_list("PROCESS WRONG\r\n", strlen("PROCESS WRONG\r\n"), wlist);
        }
        else
        {
            char buffer[BUFFER_LENGTH] = {0};
            sprintf(buffer, "SCOUNT: %d\r\n", count);
            insert_list(buffer, strlen(buffer), wlist);
        }

        break;
    }

    case END:
    {
        // LOG("command erro\n");
        insert_list("command erro\r\n", strlen("command erro\r\n"), wlist);
    }
    }
}

int kvs_request(content_node *rlist, content_node *wlist)
{
    content_node *node = rlist->next;
    while (node != rlist)
    {
        char *msg = node->data;
        char *tokens[MAX_TOKENS] = {NULL};
        int count = kvs_split_token(msg, tokens);
        kvs_parse_protocol(wlist, tokens, count);
        node = node->next;
    }
}

int init_engine()
{

#if RBTREE_ENGINE_ENABLE
    LOG("rbtree\n");
    kvs_rbtree_create(&global_rbtree);
#endif
#if HASH_ENGINE_ENABLE
    LOG("hash\n");
    global_hash_table = NULL;
    kvs_hash_create(&global_hash_table, HASH_BUCKET_COUNT);
#endif
#if SKIPLIST_ENGINE_ENABLE
    LOG("skiplist\n");
    global_skiplist = NULL;
    kvs_skiplist_create(&global_skiplist);
#endif

#if (AOF_PERSIST && INIT_BY_AOF)
    int res = aof_replay(AOF_FILE_NAME, 0, global_rbtree, global_hash_table, global_skiplist);
    if (res == 0)
    {
        LOG("增量持久化成功\n");
    }
    else
    {
        LOG("增量持久化失败\n");
    }
#endif

#if MASTER_SLAVE
    // 初始化主从配置（传递红黑树实例）
    if (ms_init(g_server_mode, g_master_ip, g_master_port, SLAVE_SYNC_PORT, global_rbtree, global_hash_table, global_skiplist) != 0)
    {
        fprintf(stderr, "主从配置初始化失败\n");
        return -1;
    }
#endif

#if AOF_PERSIST
    global_aof_writer = aof_async_writer_init(AOF_FILE_NAME, 3000);
#endif
}

int delete_engine()
{
#if MASTER_SLAVE
    ms_stop(); // 停止主从同步线程,必须放在开头，主机线程终止后还会触发一次SIGINT
#endif

#if RBTREE_ENGINE_ENABLE

    kvs_rbtree_destroy(global_rbtree);
#endif
#if HASH_ENGINE_ENABLE

    kvs_hash_destroy(global_hash_table);
#endif
#if SKIPLIST_ENGINE_ENABLE
    kvs_skiplist_destroy(global_skiplist);
#endif

#if AOF_PERSIST
    aof_async_writer_destroy(global_aof_writer);
#endif

#if AOF_CLEAR
    clear_file(AOF_FILE_NAME);
#endif
}

// 信号处理函数：当收到SIGINT时调用
void handle_sigint(int sig)
{
    // 注意：信号处理函数中应尽量使用异步安全函数（如write），printf理论上不绝对安全
    // 但简单场景下可使用
    printf("\n收到Ctrl+C中断（信号编号：%d）\n", sig);
    printf("正在执行清理操作...\n");

    delete_engine();
    // kvs_rbtree_destroy(global_rbtree);

    printf("清理完成，程序退出\n");

#if SHOW_JEMALLOC_USE_STATS    
    kvs_jemalloc_print_stats();
#endif
    exit(0); // 退出程序
}

void build_key_file()
{
    createFileIfNotExist(AOF_FILE_NAME);
    createFileIfNotExist(RB_FULL_FILE_NAME);
    createFileIfNotExist(HASH_FULL_FILE_NAME);
    createFileIfNotExist(SKIPLIST_FULL_FILE_NALE);
}

int main(int argc, char *argv[])
{
    build_key_file();

    // 新增：解析主从模式参数
#if MASTER_SLAVE
    atomic_store(&aof_file_sync_length, get_file_size(AOF_FILE_NAME));
    for (int i = 1; i < argc; i++)
    {
        if (strcmp(argv[i], "--master") == 0)
        {
            g_server_mode = MODE_MASTER;
            printf("启用主模式，监听同步端口 %d\n", SLAVE_SYNC_PORT);
        }
        else if (strcmp(argv[i], "--slave") == 0)
        {
            g_server_mode = MODE_SLAVE;
            if (i + 1 >= argc)
            {
                fprintf(stderr, "用法：%s --slave 主IP:主同步端口\n", argv[0]);
                return -1;
            }
            // 解析 master_ip:port
            char *colon = strchr(argv[i + 1], ':');
            if (!colon)
            {
                fprintf(stderr, "用法：%s --slave 主IP:主同步端口\n", argv[0]);
                return -1;
            }
            strncpy(g_master_ip, argv[i + 1], colon - argv[i + 1]);
            g_master_port = atoi(colon + 1);
            printf("启用从模式，连接主节点 %s:%d\n", g_master_ip, g_master_port);
            i++; // 跳过端口参数
        }
    }
#endif

    if (signal(SIGINT, handle_sigint) == SIG_ERR)
    {
        perror("无法注册信号处理函数");
        exit(1);
    }

    // 忽略SIGPIPE信号（不会再因EPIPE导致程序终止）
    signal(SIGPIPE, SIG_IGN); // 或使用sigaction更安全

    // 数据引擎
    init_engine();
#if MASTER_SLAVE
    // 启动主从同步线程（在引擎初始化后，网络框架启动前）
    if (ms_start() != 0)
    {
        fprintf(stderr, "主从同步启动失败\n");
        delete_engine();
        return -1;
    }
#endif

// 网络框架
#if SERVER == REACTOR
    reactor_entry(FOR_CLIENT_PORT);
#elif SERVER == PROACTOR
    proactor_entry(FOR_CLIENT_PORT);
#elif SERVER == NTY_SERVER
    nty_entry(FOR_CLIENT_PORT);
#endif

    delete_engine();
}